package com.example.tcp.server.mqtt;

import com.example.tcp.server.Client;
import com.example.tcp.server.Frame;
import io.netty.channel.Channel;
import io.netty.handler.codec.mqtt.*;
import io.netty.util.AsciiString;
import lombok.extern.slf4j.Slf4j;

import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.stream.Collectors;

import static io.netty.channel.ChannelFutureListener.CLOSE_ON_FAILURE;

@Slf4j
public class MqttClient extends Client {

    protected MqttClient(AsciiString deviceId, Channel channel) {
        super(deviceId, channel);
    }

    @Override
    protected void onRecvFrame(Frame frame) {
        MqttFrame mqttFrame = (MqttFrame)frame;

        MqttMessage msg = mqttFrame.getMessage();

        MqttMessageType messageType = msg.fixedHeader().messageType();
        log.debug("Received MQTT message, type: {}, channel: {}", messageType, channel);
        switch (messageType) {
            case CONNECT:
                processConnect((MqttConnectMessage) msg);
                break;
            case SUBSCRIBE:
                processSubscribe((MqttSubscribeMessage) msg);
                break;
            case UNSUBSCRIBE:
                processUnsubscribe((MqttUnsubscribeMessage) msg);
                break;
            case PUBLISH:
                processPublish((MqttPublishMessage) msg);
                break;
//            case PUBREC:
//                processPubRec(msg);
//                break;
//            case PUBCOMP:
//                processPubComp(msg);
//                break;
            case PUBREL:
                processPubRel(msg);
                break;
            case DISCONNECT:
                processDisconnect(msg);
//                break;
//            case PUBACK:
//                processPubAck(msg);
//                break;
            case PINGREQ:
                MqttMessage pingRespMessage = MessageUtil.getPingRespMessage();
                this.send(pingRespMessage).addListener(CLOSE_ON_FAILURE);
                break;
            default:
                log.error("Unknown MessageType: {}, channel: {}", messageType, this.channel);
                break;
        }
    }

    @Override
    public void onClientDisconnect() {

    }

    @Override
    public Integer incrementAndGetSeqId() {
        this.seqId.compareAndSet(0xffff, 1);
        return this.seqId.getAndIncrement();
    }


    protected void processConnect(MqttConnectMessage msg) {
        MqttConnectReturnCode returnCode = MqttConnectReturnCode.CONNECTION_ACCEPTED;
        MqttConnAckMessage ackMessage = MessageUtil.getConnectAckMessage(returnCode,msg.variableHeader().isCleanSession());
        this.send(ackMessage);
    }

    protected void processSubscribe(MqttSubscribeMessage msg) {
        int messageId = msg.variableHeader().messageId();
        List<Integer> ackQos = msg.payload().topicSubscriptions().stream().map(ts -> {
            return ts.qualityOfService().value();
        }).collect(Collectors.toList());
        MqttMessage subAckMessage = MessageUtil.getSubAckMessage(messageId,ackQos);
        this.send(subAckMessage);
    }

    protected void processDisconnect(MqttMessage mqttMessage) {
        this.channel.close();
    }

    protected void processUnsubscribe(MqttUnsubscribeMessage msg) {
        MqttUnsubAckMessage unsubAckMessage = MessageUtil.getUnSubAckMessage(MessageUtil.getMessageId(msg));
        this.send(unsubAckMessage);
    }

    protected void processPublish(MqttPublishMessage msg) {
        log.info("recv publish, topic: {}, payload: {}", msg.variableHeader().topicName(), msg.payload().toString(StandardCharsets.UTF_8));

        MqttQoS qos = msg.fixedHeader().qosLevel();

        switch (qos) {
            case AT_MOST_ONCE:
                break;
            case AT_LEAST_ONCE:
                MqttPubAckMessage pubAckMessage = MessageUtil.getPubAckMessage(msg.variableHeader().packetId());
                this.send(pubAckMessage);
                break;
            case EXACTLY_ONCE:
                MqttMessage pubRecMessage = MessageUtil.getPubRecMessage(msg.variableHeader().packetId());
                this.send(pubRecMessage);
                break;
        }
    }


    protected void processPubRel(MqttMessage msg) {
        int messageId = MessageUtil.getMessageId(msg);
        MqttMessage pubComMessage = MessageUtil.getPubComMessage(messageId);
        this.send(pubComMessage);
    }
}
